package com.lxb.demo3

import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

/**
 * @author: albert
 * @date: 2021/5/27 14:15
 * @description:
 */
object TableApiTest2 {
  def main(args: Array[String]): Unit = {

    // 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    // 创建模拟数据
    val dataStream: DataStream[Row] = env.fromElements(
      Row.of("tom", Int.box(12)),
      Row.of("sam", Int.box(10)),
      Row.of("tom", Int.box(10)),
      Row.of("tom", Int.box(10)),
      Row.of("sam", Int.box(10)))(Types.ROW(Types.STRING, Types.INT))


    // 设置数据格式到表格
    val inputTable = tableEnv.fromDataStream(dataStream).as("name", "score")
//val inputTable =tableEnv.fromDataStream(dataStream,$"name",$"age")
    // 创建表
    tableEnv.createTemporaryView("InputTable",inputTable)

    // 查询使用表
    val resultTable= tableEnv.sqlQuery("select name,sum(score) from InputTable group by name")

    val resultData: DataStream[Row] = tableEnv.toChangelogStream(resultTable)

    resultData.print()

    env.execute()

  }
}
